package com.yb.socket.listener;

import com.alibaba.fastjson.JSONObject;
import com.yb.socket.service.WrappedChannel;
import com.yb.socket.service.server.Server;
import com.yb.socket.service.server.ServerContext;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;

/**
 * @author daoshenzzg@163.com
 * @date 2019/1/4 11:17
 */
public class DefaultMqttMessageEventListener implements MessageEventListener {
    private static final Logger logger = LoggerFactory.getLogger(DefaultMqttMessageEventListener.class);

    @Override
    public EventBehavior channelRead(ChannelHandlerContext ctx, WrappedChannel channel, Object msg) {
        if (msg instanceof MqttMessage) {
            MqttMessage message = (MqttMessage) msg;
            MqttMessageType messageType = message.fixedHeader().messageType();
            switch (messageType) {
                case CONNECT:
                    this.connect(channel, (MqttConnectMessage) message);
                    break;
                case PUBLISH:
//                    this.publish(channel, (MqttPublishMessage) message);
                    MqttPublishMessage msg1 = (MqttPublishMessage) message;
                    ByteBuf buf = msg1.content().duplicate();
                    byte[] tmp = new byte[buf.readableBytes()];
                    buf.readBytes(tmp);
                    String content = null;
                    try {
                        content = new String(tmp,"UTF-8");
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    //校验传入的数据是否符合要求
                    if(StringUtils.isBlank(content)){
                        logger.error("MQTT接收到的数据包为空===》{}",content);
                        puback(channel,msg1,"MQTT接收到的数据包为空");
                    }
                    // 校验接收的数据是否是JSON格式
                    if(!isJsonObject(content)){
                        logger.error("MQTT接收到的数据包不为JSON格式===》{}",content);
                        puback(channel,msg1,"MQTT接收到的数据包不为JSON格式");
                    }
                    logger.info("MQTT读取到的客户端发送信息===>{}",content);
                    break;
                case SUBSCRIBE:
                    this.subscribe(channel, (MqttSubscribeMessage) message);
                    break;
                case UNSUBSCRIBE:
                    this.unSubscribe(channel, (MqttUnsubscribeMessage) message);
                    break;
                case PINGREQ:
                    this.pingReq(channel, message);
                    break;
                case DISCONNECT:
                    this.disConnect(channel, message);
                    break;
                default:
                    if (logger.isDebugEnabled()) {
                        logger.debug("Nonsupport server message  type of '{}'.", messageType);
                    }
                    break;
            }
        }
        return EventBehavior.CONTINUE;
    }
    protected String userName = "jia";
    protected String password = "123456";

    public void connect(WrappedChannel channel, MqttConnectMessage msg) {
        if (logger.isDebugEnabled()) {
            String clientId = msg.payload().clientIdentifier();
            logger.debug("MQTT CONNECT received on channel '{}', clientId is '{}'.",
                    channel.id().asShortText(), clientId);
        }
        String userName = msg.payload().userName();
        String password = msg.payload().password();

        if(!this.userName.equals(userName) || !this.password.equals(password)){
            logger.info("账号密码不正确，通道关闭===>{}:{}",userName,password);
            // 向客户端返回账号密码不正确消息，其实客户端有重连机制，可直接关闭
//            MqttConnAckMessage okResp = (MqttConnAckMessage) MqttMessageFactory.newMessage(new MqttFixedHeader(
//                            MqttMessageType.CONNACK, false, MqttQoS.AT_LEAST_ONCE, false, 0),
//                    new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, true), null);
//            channel.writeAndFlush(okResp);
            disConnect(channel,msg);
            return;
        }
        MqttConnAckMessage okResp = (MqttConnAckMessage) MqttMessageFactory.newMessage(new MqttFixedHeader(
                        MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, true), null);
        channel.writeAndFlush(okResp);
    }

    public void pingReq(WrappedChannel channel, MqttMessage msg) {
        if (logger.isDebugEnabled()) {
            logger.debug("MQTT pingReq received.");
        }

        Server server = ServerContext.getContext().getServer();
        if(server != null) {
            server.getCountInfo().getHeartbeatNum().incrementAndGet();
        }

        MqttMessage pingResp = new MqttMessage(new MqttFixedHeader(MqttMessageType.PINGRESP, false,
                MqttQoS.AT_MOST_ONCE, false, 0));
        channel.writeAndFlush(pingResp);
    }

    public void disConnect(WrappedChannel channel, MqttMessage msg) {
        if (channel.isActive()) {
            channel.close();

            if (logger.isDebugEnabled()) {
                logger.debug("MQTT channel '{}' was closed.", channel.id().asShortText());
            }
        }
    }

    public void publish(WrappedChannel channel, MqttPublishMessage msg) {
//        ByteBuf buf = msg.content().duplicate();
//        byte[] tmp = new byte[buf.readableBytes()];
//        buf.readBytes(tmp);
//        String content = null;
//        try {
//            content = new String(tmp,"UTF-8");
//        } catch (UnsupportedEncodingException e) {
//            e.printStackTrace();
//        }
//        //校验传入的数据是否符合要求
//        if(StringUtils.isBlank(content)){
//            logger.error("MQTT接收到的数据包为空===》{}",content);
//            puback(channel,msg,"MQTT接收到的数据包为空");
//            return;
//        }
//        // 校验接收的数据是否是JSON格式
//        if(!isJsonObject(content)){
//                logger.error("MQTT接收到的数据包不为JSON格式===》{}",content);
//                puback(channel,msg,"MQTT接收到的数据包不为JSON格式");
//                return;
//        }
//        logger.info("MQTT读取到的客户端发送信息===>{}",content);
    }

    /**
     * 判断字符串是否可以转化为json对象
     * @param content
     * @return
     */
    public static boolean isJsonObject(String content) {
        if(StringUtils.isBlank(content))
            return false;
        try {
            JSONObject jsonStr = JSONObject.parseObject(content);
            return true;
        } catch (Exception e) {
            return false;
        }
    }
    // 客户端QOS1消息类型( MqttQoS.AT_LEAST_ONCE = qos1)，需要服务器响应包
    private void puback(Channel channel, MqttPublishMessage msg, String payLoad){
        MqttPubAckMessage sendMessage = (MqttPubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(msg.variableHeader().packetId()),
                payLoad);
        channel.writeAndFlush(sendMessage);
    }
    public void subscribe(WrappedChannel channel, MqttSubscribeMessage msg) {
        MqttSubAckMessage subAckMessage = (MqttSubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()),
                new MqttSubAckPayload(0));
        channel.writeAndFlush(subAckMessage);
    }

    public void unSubscribe(WrappedChannel channel, MqttUnsubscribeMessage msg) {
        MqttUnsubAckMessage unSubAckMessage = (MqttUnsubAckMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()), null);
        channel.writeAndFlush(unSubAckMessage);
    }

}
